package com.kafka.model;

import com.kafka.entity.Message;
import com.kafka.kafka.KafkaProducer;
import com.kafka.logs.SystemLogger;
import com.kafka.utils.SpringContextUtil;
import com.kafka.utils.UUIDUtils;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.stereotype.Service;

import java.util.concurrent.ConcurrentHashMap;

/**
 * 网络接收数据处理类
 *
 * @author
 */
public class TcpServerHandler extends ChannelInboundHandlerAdapter {

    private TcpServerConfig tcpServerConfig;
    private ConcurrentHashMap<String, Channel> channelMap;
    private String uuid = null;
    private KafkaProducer kafkaProducer= (KafkaProducer) SpringContextUtil.getBean("kafkaProducer");

    public TcpServerHandler(TcpServerConfig tcpServerConfig, ConcurrentHashMap<String, Channel> channelMap) {
        this.tcpServerConfig = tcpServerConfig;
        this.channelMap = channelMap;
    }

    /**
     * 连接激活
     *
     * @param ctx
     * @throws Exception
     */

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        SystemLogger.info("accept terminal,ip:{}", ctx.channel().remoteAddress().toString());
    }

    /**
     * 接收消息
     *
     * @param ctx
     * @param msg
     * @throws Exception
     */
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        //设置交易报文属性
        Message message = new Message();
        message.setId(UUIDUtils.id());
        uuid = message.getId();
        String addr = ctx.channel().remoteAddress().toString()
                .replaceAll("/", "").split(":")[0];
        message.setAddr(addr);
        message.setReceiceTime(String.valueOf(System.currentTimeMillis()));
        //接收未处理
        message.setStatus("0");
//        //设置渠道ID
//        message.setChannelId(tcpServerConfig.getChannelid());
//        //设置请求交易
//        message.setbReq(true);
        //设置生命周期
        message.setTxnLife(tcpServerConfig.getTimeoutclient());
        //映射网络ID
        channelMap.put(message.getId(), ctx.channel());
        //发送至Kafka
        byte[] recerveMsgByte = (byte[]) msg;

        message.setMsg(recerveMsgByte);
        SystemLogger.info("received message,ip:{},uuid:{},messageLen:{}",
                message.getAddr(), message.getId(), message.getMsg().length);
        kafkaProducer.send(message);
    }

    /**
     * 连接关闭
     *
     * @param ctx
     */
    @Override
    public void channelInactive(ChannelHandlerContext ctx) {
        //断开连接时候删除对应KEY，清理KAFKA回复队列未收到渠道的回复报文时Map中的残留数据
        if (uuid != null) {
            if (channelMap.get(uuid) != null) {
                channelMap.remove(uuid);
            }
        }
        SystemLogger.info("diconnected terminal,ip:{}", ctx.channel().remoteAddress().toString());
    }

    /**
     * 异常处理
     *
     * @param ctx
     * @param cause
     * @throws Exception
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        super.exceptionCaught(ctx, cause);
        ctx.close();// 直接关闭channel
        SystemLogger.info("network exception,ip:{},info:{}", ctx.channel().remoteAddress().toString(), cause.getMessage());
    }
}
